package org.apache.flink.table.catalog.hive.client;

import org.apache.flink.shaded.guava18.com.google.common.base.Throwables;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.api.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;

/**
 * Wrapper class for Hive Metastore Client, which embeds a HiveShim layer to handle different Hive
 * versions. Methods provided mostly conforms to IMetaStoreClient interfaces except those that
 * require shims.
 */
public class CustomHiveClient {

    private static final Logger LOG = LoggerFactory.getLogger(CustomHiveClient.class);

    private static final String DRIVER_NAME = "org.apache.hive.jdbc.HiveDriver";

    // systemProperty key
    private static final String JAVA_SECURITY_KRB_5_CONF = "java.security.krb5.conf";

    // zookeeper节点ip和端口列表
    private static String auth = "KERBEROS";
    private static String principal = "hive/hadoop.hadoop.com@HADOOP.COM";
    private static String userName = "hive/hadoop.hadoop.com@HADOOP.COM";
    private static String krb5File = "/etc/krb5.conf";
    private static String keytabFile =
            "/opt/huawei/Bigdata/FusionInsight_HD_6.5.1/install/FusionInsight-Hive-3.1.0/hive-3.1.0/bin/hive.keytab";

    private static String hiveJdbc = "192.168.12.11:21066";

    // SQL
    public static final String SHOW_DATABASES = "SHOW DATABASES";
    public static final String CREATE_FUNCTION = "CREATE FUNCTION `?`.`?` AS '?';";


    static {
        try {
            Class.forName(DRIVER_NAME);
        } catch (ClassNotFoundException e) {
            LOG.error("HiveDriver load error ,{}", Throwables.getStackTraceAsString(e));
        }
    }

    public CustomHiveClient(HiveConf hiveConf) {
        // kerberos 配置
        principal = hiveConf.get("hive.server2.authentication.kerberos.principal");
        userName = hiveConf.get("hive.server2.authentication.kerberos.principal");
        keytabFile = hiveConf.get("hive.server2.authentication.kerberos.keytab");
        auth = hiveConf.get("hive.server2.authentication");
        krb5File = hiveConf.get("custom.krb5File");
        //
        hiveJdbc = hiveConf.get("custom.hiveJdbc");
    }

    public static List<String> getAllDatabases() {
        List<String> result = new ArrayList<>();
        Connection conn = null;
        Statement st = null;
        ResultSet rs = null;
        try {
            conn = getConnection();
            st = conn.createStatement();
            rs = st.executeQuery(SHOW_DATABASES);
            while (rs.next()) {
                result.add(rs.getString(1));
            }
        } catch (Exception e) {
            LOG.error("getAllDatabases error ,{}", Throwables.getStackTraceAsString(e));
        } finally {
            close(conn, st, rs);
        }
        return result;
    }

    public void close() {
        // DO NOTHING
    }

    private static Connection getConnection() {
        // 拼接JDBC URL
        try {
            return DriverManager.getConnection(buildURL(), "", "");
        } catch (Exception e) {
            LOG.error("getConnection error ,{}", Throwables.getStackTraceAsString(e));
        }
        return null;
    }

    private static String buildURL() {
        StringBuilder strBuilder = new StringBuilder("jdbc:hive2://").append(hiveJdbc).append("/");
        if ("KERBEROS".equalsIgnoreCase(auth)) {
            strBuilder
                    .append(";auth=")
                    .append(auth)
                    .append(";principal=")
                    .append(principal)
                    .append(";user.principal=")
                    .append(userName)
                    .append(";user.keytab=")
                    .append(keytabFile)
                    .append(";");
        }
        System.setProperty(JAVA_SECURITY_KRB_5_CONF, krb5File);
        LOG.info("HIVE_JDBC_URL:{}", strBuilder.toString());
        return strBuilder.toString();
    }


    public static void createFunction(Function function) {
        Connection conn = null;
        PreparedStatement st = null;
        ResultSet rs = null;
        try {
            conn = getConnection();
            st = conn.prepareStatement(CREATE_FUNCTION);
            st.setString(1,function.getDbName());
            st.setString(2,function.getFunctionName());
            st.setString(3,function.getClassName());
            rs = st.executeQuery();
        } catch (Exception e) {
            LOG.error("getAllDatabases error ,{}", Throwables.getStackTraceAsString(e));
        } finally {
            close(conn, st, rs);
        }
    }

    private static void close(Connection conn, Statement st, ResultSet rs) {
        if (rs != null) {
            try {
                rs.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        if (st != null) {
            try {
                st.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        if (conn != null) {
            try {
                conn.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
